package com.yinhai.uid.impl;

import com.yinhai.uid.BitsAllocator;
import com.yinhai.uid.UidGenerator;
import com.yinhai.uid.exception.UidGenerateException;
import com.yinhai.uid.utils.DateUtils;
import com.yinhai.uid.utils.NamingThreadFactory;
import com.yinhai.uid.worker.WorkerIdAssigner;
import com.yinhai.uid.zookeeper.ZookeeperRegistryCenter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class DefaultUidGenerator implements UidGenerator, InitializingBean, DisposableBean {

    private static final Logger logger = LoggerFactory.getLogger(DefaultUidGenerator.class);

    private static final String SCHEDULE_NAME = "Report-SysTime-Schedule";

    private static final String FOREVER_PATH = "/forever";

    private static final String ZK_PATH_SEPARATOR = "/";

    @Value("${uid.worker-id-assigner.zookeeper.zk-node-registry-name}")
    private String zkNodeRegistryName;

    protected int timeBits = 41;

    protected int workerBits = 10;

    protected int seqBits = 12;

    protected String epochStr = "2016-05-20";

    protected long epochSeconds = 1463673600000L;

    private long scheduleInterval;

    private long allowTimeOffset;

    protected BitsAllocator bitsAllocator;

    protected long workerId;

    protected long sequence = 0L;

    protected long lastSecond = -1L;

    protected String serverPath;

    protected WorkerIdAssigner workerIdAssigner;

    private ScheduledExecutorService sysTimeReportSchedule;

    @Autowired
    private ZookeeperRegistryCenter zookeeperRegistryCenter;

    @Override
    public void afterPropertiesSet() throws Exception {
        // 初始化位数指定器
        bitsAllocator = new BitsAllocator(timeBits, workerBits, seqBits);

        // 初始化workerId
        workerId = workerIdAssigner.assignWorkerId();
        if (workerId > bitsAllocator.getMaxWorkerId()) {
            throw new RuntimeException("Worker id " + workerId + " 超过最大值 " + bitsAllocator.getMaxWorkerId());
        }

        // 设置当前服务器在zk节点上的注册名称
        List<String> foreverNodes = zookeeperRegistryCenter.getChildrenKeys(FOREVER_PATH);
        for (String node : foreverNodes) {
            if (node.startsWith(zkNodeRegistryName)) {
                serverPath = FOREVER_PATH + ZK_PATH_SEPARATOR + node;
                break;
            }
        }
        // 每隔一段时间(时间自定义)上报自身系统时间，写入zk节点/uid/forever/${self}
        sysTimeReportSchedule = Executors.newSingleThreadScheduledExecutor(new NamingThreadFactory(SCHEDULE_NAME));
        sysTimeReportSchedule.scheduleWithFixedDelay(() -> reportSysTime(), scheduleInterval, scheduleInterval, TimeUnit.MILLISECONDS);

        logger.info("bits(1, {}, {}, {})，workerId:{}", timeBits, workerBits, seqBits, workerId);
    }

    @Override
    public void destroy() throws Exception {
        if (sysTimeReportSchedule != null && !sysTimeReportSchedule.isShutdown()) {
            sysTimeReportSchedule.shutdownNow();
        }
    }

    @Override
    public long getUID() throws UidGenerateException {
        try {
            return nextId();
        } catch (Exception e) {
            logger.error("生成单个序列号发生异常", e);
            throw new UidGenerateException(e);
        }
    }

    @Override
    public long[] getBatchUID(int number) throws UidGenerateException {
        try {
            if (number < 1) {
                throw new UidGenerateException("批量获取序列号时，数量小于1");
            }
            // 集群中当前节点系统时间和其余节点在zookeeper顺序节点/uid/forever/{self}存储的时间的平均值作比较
            // 如果【sum(time)/nodeSize - 当前系统时间】在阈值允许的范围内，则等待时间误差并获取序列号，否则抛出异常
            long currentSecond = getCurrentSecond();
            long sum = 0L;
            List<String> list = zookeeperRegistryCenter.getChildrenKeys(FOREVER_PATH);
            for (String nodeName : list) {
                sum += Long.parseLong(zookeeperRegistryCenter.get(FOREVER_PATH + ZK_PATH_SEPARATOR + nodeName));
            }
            long average = sum / list.size();
            if (currentSecond < average) {
                long offset = average - currentSecond;
                if (offset > 0L && offset <= allowTimeOffset) {
                    wait(offset);
                } else if (offset > allowTimeOffset) {
                    throw new UidGenerateException("机器时间发生了"+ offset + "毫秒长回拨");
                }
            }
            // 开始生成序列号
            long[] ids = new long[number];
            for (int i = 0; i < number; i++) {
                ids[i] = nextId();
            }
            return ids;
        } catch (Exception e) {
            logger.error("生成批量序列号发生异常", e);
            throw new UidGenerateException(e);
        }
    }

    /**
     * 获取序列号
     *
     * @return UID
     * @throws UidGenerateException
     */
    protected synchronized long nextId() {
        long currentSecond = getCurrentSecond();
        // 时钟向后移动，拒绝生成uid
        if (currentSecond < lastSecond) {
            long refusedSeconds = lastSecond - currentSecond;
            throw new UidGenerateException("时钟倒退. %d毫秒内拒绝生成序列号", refusedSeconds);
        }

        // 同一时间内, 序列号递增直到最大值
        if (currentSecond == lastSecond) {
            sequence = (sequence + 1) & bitsAllocator.getMaxSequence();
            // 超过最大值, 等待下一毫秒
            if (sequence == 0) {
                currentSecond = getNextSecond(lastSecond);
            }

        // 时间发生变化, 序列号从零开始计算
        } else {
            sequence = 0L;
        }

        lastSecond = currentSecond;

        // 分配位数生成序列号
        return bitsAllocator.allocate(currentSecond - epochSeconds, workerId, sequence);
    }

    // 定时上报系统时间到zookeeper永久顺序节点'forever'下
    private void reportSysTime() {
        zookeeperRegistryCenter.update(serverPath,String.valueOf(System.currentTimeMillis()));
        System.out.println(zookeeperRegistryCenter.get(serverPath));
    }

    /**
     * 获取下一毫秒
     */
    private long getNextSecond(long lastTimestamp) {
        long timestamp = getCurrentSecond();
        while (timestamp <= lastTimestamp) {
            timestamp = getCurrentSecond();
        }
        return timestamp;
    }

    /**
     * 获取当前系统时间
     */
    private long getCurrentSecond() {
        long currentSecond = System.currentTimeMillis();
        if (currentSecond - epochSeconds > bitsAllocator.getMaxDeltaSeconds()) {
            throw new UidGenerateException("当前系统时间超过允许范围. 拒绝生成序列号. 当前系统时间: " + currentSecond);
        }

        return currentSecond;
    }

    public void setWorkerIdAssigner(WorkerIdAssigner workerIdAssigner) {
        this.workerIdAssigner = workerIdAssigner;
    }

    public void setTimeBits(int timeBits) {
        if (timeBits > 0) {
            this.timeBits = timeBits;
        }
    }

    public void setWorkerBits(int workerBits) {
        if (workerBits > 0) {
            this.workerBits = workerBits;
        }
    }

    public void setSeqBits(int seqBits) {
        if (seqBits > 0) {
            this.seqBits = seqBits;
        }
    }

    public void setEpochStr(String epochStr) {
        if (StringUtils.isNotBlank(epochStr)) {
            this.epochStr = epochStr;
            this.epochSeconds = DateUtils.parseByDayPattern(epochStr).getTime();
        }
    }

    public void setScheduleInterval(Long scheduleInterval) {
        this.scheduleInterval = scheduleInterval;
    }

    public void setAllowTimeOffset(long allowTimeOffset) {
        this.allowTimeOffset = allowTimeOffset;
    }
}
